package com.exd.kafka.controller;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.time.LocalDateTime;
import com.alibaba.fastjson2.JSONObject;
import com.exd.kafka.clickhouse.ClickhouseServiceImpl;
import com.exd.kafka.service.MockVdataVsw;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.env.Environment;
import org.springframework.web.bind.annotation.*;
import com.exd.kafka.service.ProducerService;

@RestController
public class TestController {

	@Autowired
	private Environment environment;

	@Autowired
	private ProducerService producerService;
	@Autowired
	private ClickhouseServiceImpl clickhouseServiceImpl;
	
	@GetMapping("/test")
	public void test() {
		clickhouseServiceImpl.getSendVswData();
	}
	
	@GetMapping("/send")
	public Object index(@RequestParam String vin) throws IOException {
		FileReader fileReader = null;
		BufferedReader bufferedReader = null;
		try {
			String kafkaTopic = environment.getProperty("spring.kafka.producer.topic-name");
			String dataFilePath = environment.getProperty("spring.kafka.producer.properties.exd.source.file");
			fileReader = new FileReader(dataFilePath);
			bufferedReader = new BufferedReader(fileReader);
			String lineDataValue = bufferedReader.readLine();
			producerService.sendMessageToKafka(kafkaTopic, lineDataValue);
		} finally {
			if(bufferedReader!=null) {
				bufferedReader.close();
			}
			if(fileReader!=null) {
				fileReader.close();
			}
		}
		return "success";
	}

	@GetMapping("/vsw")
	public Object sendVsw(@RequestParam(value = "vin") String vin,
						  @RequestParam(value = "second") int second,
						  @RequestParam(value = "cycle", defaultValue = "10") int cycle,
						  @RequestParam(value = "interval", defaultValue = "0") int interval) {
		try {
			String kafkaTopic = environment.getProperty("spring.kafka.producer.topic-name");
			String path = environment.getProperty("spring.kafka.producer.properties.exd.source.path");

			Long milliSecond = MockVdataVsw.getLocalDateTimeSecond(LocalDateTime.now()) - second;
			String data = MockVdataVsw.mockVswString(path, milliSecond, second, cycle, interval);

			String[] vinList = vin.split(",");
			for (int i = 0; i < vinList.length; i++) {
				JSONObject jsonObject = new JSONObject();
				jsonObject.put("collectionTime", milliSecond * 1000);
				jsonObject.put("data", data);
				jsonObject.put("vin", vinList[i]);
				producerService.sendMessageToKafka(kafkaTopic, jsonObject.toJSONString());
			}
		} catch (Exception e) {
			throw new RuntimeException(e);
		}
		return "success";
	}

	@GetMapping("/offline")
	public Object offline(@RequestParam String identifier) throws IOException {
		JSONObject JSONObject = new JSONObject();
		JSONObject.put("identifier", identifier);
		producerService.sendMessageToKafka("exd_offline", JSONObject.toJSONString());
		return "success";
	}

	@PostMapping("/sendJson")
	public Object sendJson(@RequestBody JSONObject jsonObject) throws IOException {
		producerService.sendMessageToKafka("exd_vsw", jsonObject.toJSONString());
		return "success";
	}

//	@PostMapping("/sendJson")
//	public Object sendJson(@RequestParam("file") MultipartFile file) throws IOException {
//		Objects.requireNonNull(file.getOriginalFilename())
//
//		producerService.sendMessageToKafka("exd_vsw", jsonObject.toJSONString());
//		return "success";
//	}
}
